package cn.dmp.image

import cn.dmp.util.{AppParams, Jpools, RedisHander}
import org.apache.spark.broadcast.Broadcast
import org.apache.spark.rdd.RDD
import org.apache.spark.sql.Row
import redis.clients.jedis.Jedis

import scala.collection.mutable.ArrayBuffer

object KpiServiceMangerHander {
  /**
    * 用户画像处理 方法一
    *
    * @param logMap
    */
  def accKpiImageInfo(logMap: RDD[(String, String, String, String, String, String, List[String], String, String)]) = {
    //过滤无效用户
    val filterInfo: RDD[(String, String, String, String, String, String, List[String], String, String)] = logMap.filter(t => !t._1.equals("-1"))

    val groupInfo: RDD[(String, Iterable[(String, String, String, String, String, String, List[String], String, String)])] = filterInfo.groupBy(t => t._1)

    val results: RDD[String] = groupInfo.map(itr => {

      //每一个用户进行统计
      val list: List[(String, String, String, String, String, String, List[String], String, String)] = itr._2.toList
      //抽取keys
      val keywords: List[List[String]] = list.map(t => t._7)
      //keywords.foreach(println)


      //压平，wordcount统计  过滤指标为-1的标签
      val keysResults: Map[String, Int] = keywords.flatMap(t => t).map(t => (t, 1)).groupBy(t => t._1).filter(t => !t._1.equals("-1")).map(t => (t._1, t._2.size))
      val keysWordResults: String = keysResults.toList.mkString(":")

      //其他指标wordcount统计
      val otherList: List[List[String]] = list.map(t => {
        //注意 t._1 是用户uid需要排除
        List[String](t._2, t._3, t._4, t._5, t._6, t._8, t._9)
      })

      //压平，wordcount统计 过滤指标为-1的标签
      val otherResults: Map[String, Int] = otherList.flatMap(t => t).map((_, 1)).groupBy(_._1).filter(t => !t._1.equals("-1")).map(t => (t._1, t._2.size))

      itr._1 + "  " + otherResults.toList.mkString(",") + keysWordResults
    })

    results.foreach(println)
  }

  /**
    * 方法二  通过redis进行用户画像的处理
    *
    * @param logMap
    */
  def accKpiImageInfoRedis(logMap: RDD[(String, String, String, String, String, String, List[String], String, String)]) = {
    logMap.foreachPartition(itr => {

      val jedis: Jedis = Jpools.getJedis(AppParams.jedisImageNum)

      itr.foreach(tp => {
        if (!tp._1.equals("-1")) {
          if (!tp._2.equals("-1")) jedis.hincrBy(tp._1, tp._2, 1)
          if (!tp._3.equals("-1")) jedis.hincrBy(tp._1, tp._3, 1)
          if (!tp._4.equals("-1")) jedis.hincrBy(tp._1, tp._4, 1)
          if (!tp._5.equals("-1")) jedis.hincrBy(tp._1, tp._5, 1)
          if (!tp._6.equals("-1")) jedis.hincrBy(tp._1, tp._6, 1)
          if (!tp._8.equals("-1")) jedis.hincrBy(tp._1, tp._8, 1)
          if (!tp._9.equals("-1")) jedis.hincrBy(tp._1, tp._9, 1)
          tp._7.foreach(t => {
            if (!t.equals("-1")) jedis.hincrBy(tp._1, t, 1)
          })
        }
      })
      jedis.close()

    })
  }

  /**
    * 用户画像参数封装
    * @param row
    * @param clientMap
    * @param networkMap
    * @param ispnameMap
    * @param sotopWordBT
    * @return
    */
  def accUserImageMapInfo(row: Row, clientMap: Map[String, AnyRef], networkMap: Map[String, AnyRef], ispnameMap: Map[String, AnyRef], sotopWordBT: Broadcast[Array[String]]) = {
    val adspacetype: Int = row.getAs[Int]("adspacetype")
    //1) 广告位类型
    var adspacetypeImage = "-1"
    if (!adspacetype.equals("0")) {
      adspacetypeImage = if (adspacetype < 10) "LC0" + adspacetype else "LC" + adspacetype
    }

    val adspacetypename: String = row.getAs[String]("adspacetypename")
    //1) 广告位类型 名称
    var adspacetypenameImage = "LN" + adspacetypename

    val appid: String = row.getAs[String]("appid")
    val appname: String = row.getAs[String]("appname")
    //redis获取appname
    //2) App 名称
    var appnameImage: String = RedisHander.getAppName(appname, appid)

    if (appnameImage.equals("0")) appnameImage = "-1" else appnameImage = "APP" + appnameImage


    val adplatformproviderid: Int = row.getAs[Int]("adplatformproviderid")
    //3) 渠道
    val adplatformprovideridImage = if (adplatformproviderid != 0) "CN" + adplatformproviderid else "-1"

    val client: Int = row.getAs[Int]("client") 
    //4) 设备 设备操作系统

    val clientImage: String = clientMap.getOrElse(client.toString, "D00010004").toString


    val networkmannername: String = row.getAs[String]("networkmannername")
    //4) 设备 设备操作系统
    val networkImage: String = networkMap.getOrElse(networkmannername, "D00020005").toString

    val ispname: String = row.getAs[String]("ispname")
    //4) 设备 设备运营商方式
    val ispnameImage: String = ispnameMap.getOrElse("ispname", "D00030004").toString

    val keywords: String = row.getAs[String]("keywords")
    //5) 关键字
    val buffer: ArrayBuffer[String] = accKeywords(keywords, sotopWordBT)

    val provincename: String = row.getAs[String]("provincename")

    //6) 地域标签（省标签格式
    val provincenameImage: String = if (!provincename.equals("0")) "ZP" + provincename else "-1"

    val cityname: String = row.getAs[String]("cityname")
    //6) 地域标签（地市签格式
    val citynameImage: String = if (!cityname.equals("0")) "ZC" + cityname else "-1"

    //需要得到 userId
    val imei: String = row.getAs[String]("imei")
    val mac: String = row.getAs[String]("mac")
    val idfa: String = row.getAs[String]("idfa")
    val openudid: String = row.getAs[String]("openudid")
    val androidid: String = row.getAs[String]("androidid")
    val imeimd5: String = row.getAs[String]("imeimd5")
    val macmd5: String = row.getAs[String]("macmd5")
    val idfamd5: String = row.getAs[String]("idfamd5")
    val openudidmd5: String = row.getAs[String]("openudidmd5")
    val androididmd5: String = row.getAs[String]("androididmd5")
    val imeisha1: String = row.getAs[String]("imeisha1")
    val macsha1: String = row.getAs[String]("macsha1")
    val idfasha1: String = row.getAs[String]("idfasha1")
    val openudidsha1: String = row.getAs[String]("openudidsha1")
    val androididsha1: String = row.getAs[String]("androididsha1")

    val userId: String = KpiServiceMangerHander.getUserId(imei, mac, idfa, openudid, androidid, imeimd5, macmd5, idfamd5, openudidmd5, androididmd5, imeisha1, macsha1, idfasha1, openudidsha1, androididsha1)

    (userId, adspacetypeImage, adspacetypenameImage, appnameImage, adplatformprovideridImage, ispnameImage, buffer.toList, provincenameImage, citynameImage)
  }

  /**
    * keywords切分处理
    *
    * @param keywords
    * @param sotopWordBT
    * @return
    */
  def accKeywords(keywords: String, sotopWordBT: Broadcast[Array[String]]): ArrayBuffer[String] = {
    val buffer: ArrayBuffer[String] = new ArrayBuffer[String]()
    val tbValue: Array[String] = sotopWordBT.value
    if (!keywords.equals("0") && keywords.length >= 3 && keywords.length <= 8) {
      if (keywords.indexOf("|") > 0) {
        val words: Array[String] = keywords.split("\\|")
        words.foreach(word => {
          if (!tbValue.contains(keywords) && keywords.length >= 3 && keywords.length <= 8) {
            buffer += "K" + word
          }
        })
      } else if (!tbValue.contains(keywords)) {
        buffer += "K" + keywords
      }
    }
    buffer
  }

  /**
    * 获取一个识别id
    *
    * @param imei
    * @param mac
    * @param idfa
    * @param openudid
    * @param androidid
    * @param imeimd5
    * @param macmd5
    * @param idfamd5
    * @param openudidmd5
    * @param androididmd5
    * @param imeisha1
    * @param macsha1
    * @param idfasha1
    * @param openudidsha1
    * @param androididsha1
    * @return
    */
  def getUserId(imei: String,
                mac: String,
                idfa: String,
                openudid: String,
                androidid: String,
                imeimd5: String,
                macmd5: String,
                idfamd5: String,
                openudidmd5: String,
                androididmd5: String,
                imeisha1: String,
                macsha1: String,
                idfasha1: String,
                openudidsha1: String,
                androididsha1: String): String = {
    if (!imei.equals("0")) {
      "D:" + imei.toUpperCase
    } else if (!mac.equals("0")) {
      "D:" + mac.toUpperCase
    } else if (!idfa.equals("0")) {
      "D:" + idfa.toUpperCase
    } else if (!openudid.equals("0")) {
      "D:" + openudid.toUpperCase
    } else if (!androidid.equals("0")) {
      "D:" + androidid.toUpperCase
    } else if (!imeimd5.equals("0")) {
      "D:" + imeimd5.toUpperCase
    } else if (!macmd5.equals("0")) {
      "D:" + macmd5.toUpperCase
    } else if (!idfamd5.equals("0")) {
      "D:" + idfamd5.toUpperCase
    } else if (!openudidmd5.equals("0")) {
      "D:" + openudidmd5.toUpperCase
    } else if (!androididmd5.equals("0")) {
      "D:" + androididmd5.toUpperCase
    } else if (!imeisha1.equals("0")) {
      "D:" + imeisha1.toUpperCase
    } else if (!macsha1.equals("0")) {
      "D:" + macsha1.toUpperCase
    } else if (!idfasha1.equals("0")) {
      "D:" + idfasha1.toUpperCase
    } else if (!openudidsha1.equals("0")) {
      "D:" + openudidsha1.toUpperCase
    } else if (!androididsha1.equals("0")) {
      "D:" + androididsha1.toUpperCase
    } else {
      "-1"
    }
  }
}
